import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
import java.util.Random;
import java.util.UUID;

/**
 * Kafka 生产者模拟程序：合并抖音 5 张 ODS 表字段，生成模拟数据发送到 Kafka
 * 覆盖表：ods_douyin_order / room / interact / traffic / product
 */
public class KafkaDouyinMergedProducer {

    // 定义合并后的业务实体（模拟宽表）
    public static class DouyinLiveBusiness {
        // 订单表字段（ods_douyin_order）
        private String orderId;          // 订单ID
        private String eventTime;        // 订单事件时间
        private String roomId;           // 直播间ID
        private String userId;           // 下单用户ID
        private String productSku;       // 商品SKU
        private double orderAmount;      // 订单原始金额
        private double payAmount;        // 实际支付金额
        private String payStatus;        // 支付状态

        // 直播间表字段（ods_douyin_room）
        private String anchorId;         // 主播ID
        private String roomType;         // 直播间类型
        private String startTime;        // 开播时间
        private String endTime;          // 关播时间

        // 互动表字段（ods_douyin_interact）
        private String interactId;       // 互动事件ID
        private String interactEventType;// 互动发生时间
        private String interactType;     // 互动类型（like/comment/share/follow）
        private String content;          // 互动内容

        // 流量表字段（ods_douyin_traffic）
        private String visitId;          // 访问会话ID
        private String enterTime;        // 进入直播间时间
        private String leaveTime;        // 离开直播间时间
        private String source;           // 流量入口
        private int duration;            // 停留时长（秒）

        // 商品表字段（ods_douyin_product）
        private String productName;      // 商品名称
        private double price;            // 商品售价
        private String category;         // 商品分类
        private String status;           // 上下架状态

        // Getter & Setter（Lombok @Data 可简化，此处为完整示例）
        public String getOrderId() { return orderId; }
        public void setOrderId(String orderId) { this.orderId = orderId; }

        public String getEventTime() {
            return eventTime;
        }

        public void setEventTime(String eventTime) {
            this.eventTime = eventTime;
        }

        public String getRoomId() {
            return roomId;
        }

        public void setRoomId(String roomId) {
            this.roomId = roomId;
        }

        public String getUserId() {
            return userId;
        }

        public void setUserId(String userId) {
            this.userId = userId;
        }

        public String getProductSku() {
            return productSku;
        }

        public void setProductSku(String productSku) {
            this.productSku = productSku;
        }

        public double getOrderAmount() {
            return orderAmount;
        }

        public void setOrderAmount(double orderAmount) {
            this.orderAmount = orderAmount;
        }

        public double getPayAmount() {
            return payAmount;
        }

        public void setPayAmount(double payAmount) {
            this.payAmount = payAmount;
        }

        public String getPayStatus() {
            return payStatus;
        }

        public void setPayStatus(String payStatus) {
            this.payStatus = payStatus;
        }

        // 直播间表字段 Getter & Setter
        public String getAnchorId() {
            return anchorId;
        }

        public void setAnchorId(String anchorId) {
            this.anchorId = anchorId;
        }

        public String getRoomType() {
            return roomType;
        }

        public void setRoomType(String roomType) {
            this.roomType = roomType;
        }

        public String getStartTime() {
            return startTime;
        }

        public void setStartTime(String startTime) {
            this.startTime = startTime;
        }

        public String getEndTime() {
            return endTime;
        }

        public void setEndTime(String endTime) {
            this.endTime = endTime;
        }

        // 互动表字段 Getter & Setter
        public String getInteractId() {
            return interactId;
        }

        public void setInteractId(String interactId) {
            this.interactId = interactId;
        }

        public String getInteractEventType() {
            return interactEventType;
        }

        public void setInteractEventType(String interactEventType) {
            this.interactEventType = interactEventType;
        }

        public String getInteractType() {
            return interactType;
        }

        public void setInteractType(String interactType) {
            this.interactType = interactType;
        }

        public String getContent() {
            return content;
        }

        public void setContent(String content) {
            this.content = content;
        }

        // 流量表字段 Getter & Setter
        public String getVisitId() {
            return visitId;
        }

        public void setVisitId(String visitId) {
            this.visitId = visitId;
        }

        public String getEnterTime() {
            return enterTime;
        }

        public void setEnterTime(String enterTime) {
            this.enterTime = enterTime;
        }

        public String getLeaveTime() {
            return leaveTime;
        }

        public void setLeaveTime(String leaveTime) {
            this.leaveTime = leaveTime;
        }

        public String getSource() {
            return source;
        }

        public void setSource(String source) {
            this.source = source;
        }

        public int getDuration() {
            return duration;
        }

        public void setDuration(int duration) {
            this.duration = duration;
        }

        // 商品表字段 Getter & Setter
        public String getProductName() {
            return productName;
        }

        public void setProductName(String productName) {
            this.productName = productName;
        }

        public double getPrice() {
            return price;
        }

        public void setPrice(double price) {
            this.price = price;
        }

        public String getCategory() {
            return category;
        }

        public void setCategory(String category) {
            this.category = category;
        }

        public String getStatus() {
            return status;
        }

        public void setStatus(String status) {
            this.status = status;
        }
        // ... 其他字段的 Getter & Setter 需补充 ...
    }

    public static void main(String[] args) {
        // Kafka 配置
        Properties props = new Properties();
        props.put("bootstrap.servers", "node101:9092,node102:9092,node103:9092"); // Kafka 地址
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        // 初始化 Kafka 生产者
        Producer<String, String> producer = new KafkaProducer<>(props);
        String topic = "douyin_live_business"; // Kafka Topic 名称

        // 随机数据生成器
        Random random = new Random();
        String[] payStatus = {"unpay", "pay_success", "refund"};
        String[] roomType = {"sales", "knowledge", "entertainment"};
        String[] interactType = {"like", "comment", "share", "follow"};
        String[] source = {"app", "h5", "share", "search"};
        String[] category = {"clothing", "food", "beauty"};
        String[] status = {"on_shelf", "off_shelf"};
         try {


             while (true) { // 持续发送数据
                DouyinLiveBusiness data = new DouyinLiveBusiness();

                // 订单表字段
                data.setOrderId(UUID.randomUUID().toString().replace("-", ""));
                data.setEventTime("2025-08-21 " + String.format("%02d:%02d:%02d", random.nextInt(24), random.nextInt(60), random.nextInt(60)));
                data.setRoomId("Room" + System.currentTimeMillis());
                data.setUserId("User" + random.nextInt(1000000));
                data.setProductSku("SKU" + System.currentTimeMillis());
                data.setOrderAmount(random.nextDouble() * 1000); // 订单金额
                data.setPayAmount(random.nextDouble() * 900);   // 支付金额
                data.setPayStatus(payStatus[random.nextInt(payStatus.length)]);

                // 直播间表字段
                data.setAnchorId("Anchor" + random.nextInt(1000));
                data.setRoomType(roomType[random.nextInt(roomType.length)]);
                data.setStartTime("2025-08-21 " + String.format("%02d:%02d:%02d", random.nextInt(24), random.nextInt(60), random.nextInt(60)));
                data.setEndTime(random.nextBoolean() ? "2025-08-21 " + String.format("%02d:%02d:%02d", random.nextInt(24), random.nextInt(60), random.nextInt(60)) : "");

                // 互动表字段
                data.setInteractId("INT" + System.currentTimeMillis());
                data.setInteractEventType("2025-08-21 " + String.format("%02d:%02d:%02d", random.nextInt(24), random.nextInt(60), random.nextInt(60)));
                data.setInteractType(interactType[random.nextInt(interactType.length)]);
                data.setContent("互动内容：" + UUID.randomUUID().toString().substring(0, 10));

                // 流量表字段
                data.setVisitId("VIS" + System.currentTimeMillis());
                data.setEnterTime("2025-08-21 " + String.format("%02d:%02d:%02d", random.nextInt(24), random.nextInt(60), random.nextInt(60)));
                data.setLeaveTime(random.nextBoolean() ? "2025-08-21 " + String.format("%02d:%02d:%02d", random.nextInt(24), random.nextInt(60), random.nextInt(60)) : "");
                data.setSource(source[random.nextInt(source.length)]);
                data.setDuration(random.nextInt(300)); // 停留时长（秒）

                // 商品表字段
                data.setProductName("商品名称_" + UUID.randomUUID().toString().substring(0, 8));
                data.setPrice(random.nextDouble() * 100); // 商品售价
                data.setCategory(category[random.nextInt(category.length)]);
                data.setStatus(status[random.nextInt(status.length)]);

                // 转为 JSON 发送到 Kafka
                String json = "{\"orderId\":\"" + data.getOrderId() + "\"," +
                        "\"eventTime\":\"" + data.getEventTime() + "\"," +
                        "\"roomId\":\"" + data.getRoomId() + "\"," +
                        "\"userId\":\"" + data.getUserId() + "\"," +
                        "\"productSku\":\"" + data.getProductSku() + "\"," +
                        "\"orderAmount\":" + data.getOrderAmount() + "," +
                        "\"payAmount\":" + data.getPayAmount() + "," +
                        "\"payStatus\":\"" + data.getPayStatus() + "\"," +
                        "\"anchorId\":\"" + data.getAnchorId() + "\"," +
                        "\"roomType\":\"" + data.getRoomType() + "\"," +
                        "\"startTime\":\"" + data.getStartTime() + "\"," +
                        "\"endTime\":\"" + data.getEndTime() + "\"," +
                        "\"interactId\":\"" + data.getInteractId() + "\"," +
                        "\"interactEventType\":\"" + data.getInteractEventType() + "\"," +
                        "\"interactType\":\"" + data.getInteractType() + "\"," +
                        "\"content\":\"" + data.getContent() + "\"," +
                        "\"visitId\":\"" + data.getVisitId() + "\"," +
                        "\"enterTime\":\"" + data.getEnterTime() + "\"," +
                        "\"leaveTime\":\"" + data.getLeaveTime() + "\"," +
                        "\"source\":\"" + data.getSource() + "\"," +
                        "\"duration\":" + data.getDuration() + "," +
                        "\"productName\":\"" + data.getProductName() + "\"," +
                        "\"price\":" + data.getPrice() + "," +
                        "\"category\":\"" + data.getCategory() + "\"," +
                        "\"status\":\"" + data.getStatus() + "\"}";

                ProducerRecord<String, String> record = new ProducerRecord<>(topic, json);
                producer.send(record);
                System.out.println("发送数据：" + json);

                // 控制发送频率（每 1 秒发送 1 条）
                Thread.sleep(1000);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }
}
